// -*-mode:c++; coding:utf-8-*-

#include "imphub_dbp_protocol.hpp"

template<typename session_ptr>
class basic_request_queue
{
public:
	typedef std::pair<session_ptr, query_request_ptr> query_request;
	typedef std::pair<session_ptr, insert_request_ptr> insert_request;
	typedef std::deque<query_request> query_queue;
	typedef std::deque<query_request> insert_queue;
	
public:
	basic_request_queue(std::size_t size)
		: m_notify_size(size)
		, m_stopped(false) {}

	bool stop() {
		boost::mutex::scoped_lock lock(m_query_mutex);
		boost::mutex::scoped_lock lock(m_insert_mutex);
		m_stopped = true;
		m_query_cond.notify_all();
		m_insert_cond.notify_all();
	}

	void push_back(session_ptr s, packet_ptr p) {
		if(p->m_type == QUERY_REQ)
		{
			query_request_ptr ptr(boost::polymorphic_downcast<query_request*>(p.get()));
			boost::mutex::scoped_lock lock(m_query_mutex);
			bool need_notify = m_query_queue.empty();
			m_query_queue.push_back(query_request(s, ptr));
			if(need_notify)
				m_query_cond.notify_one();
		}
		else if(p->m_type == INSERT_REQ)
		{
			insert_request_ptr ptr(boost::polymorphic_downcast<insert_request*>(p.get()));
			boost::mutex::scoped_lock lock(m_insert_mutex);
			bool need_notify = m_insert_queue.size() < m_notify_size;
			m_insert_queue.push_back(insert_request(s, ptr));
			if(m_queue.size() >= m_notify_size)
				m_insert_cond.notify_one();
		}
		else
		{
			LOG(ERROR, "bad request type, packet = " << ptr->to_string());
		}
	}

	bool get_query(query_queue &q) {
		boost::mutex::scoped_lock lock(m_query_mutex);
		while((! m_stopped)
		      && m_query_queue.empty())
		{
			m_query_cond.wait(lock);
		}
		if(! m_query_queue.empty())
		{
			q.swap(m_query_queue);
		}
		return (! m_stopped);
	}

	bool get_insert(insert_queue &q) {
		boost::mutex::scoped_lock lock(m_insert_mutex);
		while((! m_stopped)
		      && (m_insert_queue.size() < m_notify_size))
		{
			m_insert_cond.wait(lock);
		}
		if(! m_insert_queue.empty())
		{
			q.swap(m_insert_queue);
		}
		return (! m_stopped);
	}

private:
	bool m_stopped;

	query_queue m_query_queue;
	boost::mutex m_query_mutex;
	boost::condition_variable m_query_cond;

	insert_queue m_insert_queue;
	boost::mutex m_insert_mutex;
	boost::condition_variable m_insert_cond;
	std::size_t m_notify_size;
};

class server_session
{
public:
	typedef server_session this_type;
	typedef boost::intrusive_ptr<server_session> ptr_type;

	typedef boost::asio::io_service io_service;
	typedef boost::asio::ip::tcp::socket socket_type;
	typedef boost::shared_ptr<socket_type> socket_ptr;
	typedef socket_type::endpoint_type endpoint_type;
	typedef boost::system::error_code error_code;

	typedef std::deque<packet_ptr> response_queue;
	typedef basic_request_queue<ptr_type> request_queue;

public:
	server_session(io_service &ios,
		       request_queue &rq)
		: m_socket(ios)
		, m_request_queue(rq) {
		atomic_set(&m_reference_counter, 0);
	}

	socket_type &socket() {
		return m_socket;
	}

	// 只能被调用一次
	void async_read() {
		active_async_read();
	}

	void async_write(const packet_ptr &ptr) {
		boost::mutex::scoped_lock lock(m_response_mutex);
		const bool need_active = m_response_queue.empty();
		m_response_queue.push_back(ptr);
		lock.unlock();
		
		if(need_active)
		{
			active_async_write();
		}
	}

	void close() {
		m_socket.close();
		m_read_buf.clear();
	}

protected:
	void active_async_read() {
		boost::asio::async_read(m_socket,
					m_read_buf.prepare(packet::buffer_size),
					boost::asio::transfer_at_least(packet::header_size),
					boost::bind(&server_session::on_read,
						    this_ptr(),
						    _1, _2));
	}

	void active_async_write() {
		boost::mutex::scoped_lock lock(m_response_mutex);
		while(! m_response_queue.empty())
		{
			packet_ptr ptr = m_response_queue.front();
			lock.unlock();

			std::size_t buffer_size = packet::buffer_size;
			bool ok = false;
			for(;;)
			{
				boost::asio::mutable_buffer buf =
					m_write_buf.prepare(buffer_size);
				oarchive out(buf);
				ok = save(out, ptr);
				if(ok)
				{
					m_write_buf.commit(out.used_buffer_size());
					break;
				}
				else
				{
					buffer_size <<= 1;
					if(buffer_size >= 1024 * 1024 * 1024) // 1G
						break;
				}
			}
			if(ok)
			{
				boost::asio::async_write(m_socket,
							 m_write_buf.data(),
							 boost::asio::transfer_all(),
							 boost::bind(&server_session::on_written,
								     this_ptr(), _1, _2));
				break;
			}

			LOG(ERROR, "save packet failed, packet = "
			    << ptr->to_string());
			
			lock.lock();
			m_response_queue.pop();
			// keep the lock and continue next
		}
	}

	void on_written(const error_code &ec,
			std::size_t bytes) {
		m_write_buf.consume(bytes);
		boost::mutex::scoped_lock lock(m_response_mutex);
		packet_ptr ptr = m_response_queue.front();
		m_response_queue.pop_front();
		lock.unlock();
		if(ec)
		{
			m_write_buf.clear();
			LOG(ERROR, "write packet failed, ec = " << ec
			    << " - " << ec.message()
			    << ", packet = " << ptr->to_string());
		}
		active_async_write();
	}

	void on_read(const error_code &e, std::size_t bytes_transferred) {
		m_read_buf.commit(bytes_transferred);
		std::size_t bytes_to_read = 0;

		error_code ec = e;
		if(! ec)
		{
			const boost::asio::const_buffer data = m_read_buf.data();
			iarchive in(data);
			for(; ! in.empty();)
			{
				ec = validate(in, &bytes_to_read);
				if(ec)
					break;

				LOG_ENV(std::size_t size = in.left_buffer_size());
				LOG_ENV(const char *data = in.left_buffer());

				packet *ptr = load(in);
				if(ptr == NULL)
				{
					LOG(ERROR, "load packet failed, bytes_transferred = "
					    << bytes_transferred << ", data size = "
					    << size << ", data = " << dump_string_hex(data, size));
					close();
					return;
				}
				m_request_queue.push_back(this_ptr(), ptr);
			}
			m_read_buffer.consume(in.used_buffer_size());
		}
		
		if(! ec)
		{
			active_async_read();
		}
		else if((bytes_to_read != 0) &&
			((ec == xbase::error::header_incomplete) ||
			 (ec == xbase::error::packet_incomplete)))
		{
			const std::size_t buffer_size = (std::max)(bytes_to_read,
								   m_read_buf.left_space());
			boost::asio::async_read(m_socket,
						m_read_buf.prepare(buffer_size),
						boost::asio::transfer_at_least(bytes_to_read),
						boost::bind(&server_session::on_read,
							    this_ptr(),
							    boost::asio::placeholders::error,
							    boost::asio::placeholders::bytes_transferred));
		}
		else
		{
			LOG_ENV(const boost::asio::const_buffer data = m_read_buf.data());
			LOG_ENV(const std::size_t size =  boost::asio::buffer_size(data));

			LOG(ERROR, "read error, data size in read buffer = " << size
			    << ", bytes_transferred = " << bytes_transferred
			    << ", data = " << xbase::dump_string_hex(data));
			
			close();
		}
	}

	ptr_type this_ptr() {
		return ptr_type(this);
	}

private:
	socket_type m_socket;
	xbase::simple_buffer m_read_buf;
	
	request_queue &m_queue;

	response_queue m_response_queue;
	xbase::simple_buffer m_write_buf;
	boost::mutex m_response_mutex;

	atomic_t m_reference_counter;

	friend inline
	void intrusive_ptr_add_ref(server_session *s) {
		atomic_inc(&s->m_reference_counter);
	}

	friend inline
	void intrusive_ptr_release(server_session *s) {
		if(atomic_dec_and_test(&s->m_reference_counter))
			::delete s;
	}
};

class imphub_dbp
{
public:
	typedef boost::asio::ip::tcp::socket socket_type;
	typedef boost::shared_ptr<socket_type> socket_ptr;
	typedef socket_type::endpoint_type endpoint_type;
	typedef boost::system::error_code error_code;

	typedef server_session::request_queue request_queue;

public:
	void run() {
		m_acceptor.open(m_listen_endpoint.protocol());
		m_acceptor.set_option(boost::asio::ip::tcp::acceptor::reuse_address(true));
		m_acceptor.bind(m_listen_endpoint);
		m_acceptor.listen();

		async_accept(new_socket());
	}

	void start_service_thread() {
	}

protected:
	server_session_ptr new_session() {
		return server_session_ptr(
			new server_session(
				m_xbase_client_env.
				get_io_service_pool().
				get_io_service<xbase::io_service_pool::user_level>(),
				m_request_queue,
				m_request_mutex
				)
			);
	}
	
	void async_accept(server_session_ptr ptr) {
		m_acceptor.async_accept(ptr->socket(),
					boost::bind(&service::handle_accept,
						    this,
						    ptr,
						    _1));
	}

	void handle_accept(server_session_ptr ptr, const error_code& ec) {
		if(ec)
		{
			LOG(ERROR, "ec = " << ec << " - " << ec.message());
			async_accept(ptr);
		}
		else
		{
			LOG_ENV(error_code ec_ignore);
			LOG(INFO, "new connection accepted: "
			    << ptr->remote_endpoint(ec_ignore)
			    << " ---> "
			    << ptr->local_endpoint(ec_ignore));
			start_service_on_socket(ptr);
			async_accept(new_session());
		}
	}

	void start_service_on_socket(server_session_ptr ptr) {
		ptr->async_read();
	}

	void insert_service() {
		LOG(INFO, "insert_service start");
		request_queue queue;
		for(;;)
		{
			const bool not_stop = m_request_queue.get_insert(queue);
			if(! not_stop)
			{
				break;
			}
			insert_data(queue);
		}
		LOG(INFO, "insert_service quit");
	}

	void insert_data(request_queue &queue) {
		typedef std::map<std::string, insert_request_ptr> req_map;
		req_map req;
		while(! queue.empty())
		{
			req[queue.front().second->m_cid] =
				queue.front().second;
			queue.pop_front();
		}

		std::vector<row_mutation_data_ptr> mutations;
		for(req_map::iterator it=req.begin();
		    it != req.end(); ++it)
		{
			if(it->second->m_cid.size() != CID_LEN)
			{
				LOG(ERROR, "bad cid");
				continue;
			}
			if(it->second->m_gcid.size() != GCID_LEN)
			{
				LOG(ERROR, "bad gcid");
				continue;
			}

			std::string value(GCID_LEN + sizeof(uint64_t), '\0');
			if(! it->second->m_gcid.empty())
			{
				memcpy((char*)value.c_str(),
				       it->second->m_gcid.c_str(),
				       it->second->m_gcid.length());
			}
			*reinterpret_cast<uint64_t*>((char*)value.c_str() + GCID_LEN) = it->second->m_filesize;

			for(std::vector<std::string>::iterator pit=pit->second->m_peerid_list.begin();
			    pit != pit->second->m_peerid_list.end();
			    ++pit)
			{
				if(pit->size() != PEERID_LEN)
				{
					LOG(ERROR, "bad peerid");
					continue;
				}

				std::string row(CID_LEN + PEERID_LEN, '\0');
				memcpy((char*)row.data(), it->second->m_cid.data(), CID_LEN);
				memcpy((char*)(row.data() + CID_LEN), pit->data(), PEERID_LEN);
				
				row_mutation m(m_table_info, row);
				m.add(m_column, value);
				mutations.push_back(m.data_ptr());
			}
		}
	}

	void query_service() {
	}

private:
	request_queue m_request_queue;

	xbase::client_environment_ptr m_xbase_client_env;
	endpoint_type m_listen_endpoint;
	boost::asio::ip::tcp::acceptor m_acceptor;

	boost::thread_group m_threads;
};
